#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>


typedef struct Task{
    void (*task_func) (void *arg);
    void *task_arg;
}Task;

typedef struct ThreadPool {
    // 任务队列
    Task *task_queue;
    int tq_capacity;
    int tq_size;
    int tq_front;
    int tq_back;

    // 管理者和工作者线程
    pthread_t managerID;
    pthread_t *workersIDs;

    // 线程参数
    int min_num;
    int max_num;
    int busy_num;
    int alive_num;
    int kill_num;
    int isShutdown;

    // 锁与条件变量
    pthread_mutex_t pool_lock;
    pthread_mutex_t busy_lock;
    pthread_cond_t tq_notFull_cond;
    pthread_cond_t tq_notEmpty_cond;
}ThreadPool;


// 管理线程与工作线程内容
void delWorker(ThreadPool *pool) {
    for (int i = 0; i < pool->max_num; ++i) {
        if (pool->workersIDs[i] == pthread_self()) {
            pool->workersIDs[i] = 0;
//            pool->alive_num -= 1;
            printf("worker ID %ld has been deleted.\n", pthread_self());
            break;
        }
    }
    pthread_exit(NULL);
}


_Noreturn void *workers_job(void *th_pool) {
    /*当任务队列不为空时， 取出和执行任务并唤醒生产者；或遇到线程池关闭、线程需要销毁时自行销毁*/
    ThreadPool *pool = (ThreadPool *)th_pool;
    printf("Worker %ld create.\n", pthread_self());

    while (1) {
        pthread_mutex_lock(&pool->pool_lock);
        while (pool->tq_size == 0 && !pool->isShutdown) {  // while避免假唤醒的不良后果
            // 工作队列为空时阻塞等待
            pthread_cond_wait(&pool->tq_notEmpty_cond, &pool->pool_lock);

            // 是否需要销毁线程
            if (pool->kill_num > 0) {
                pool->kill_num -= 1;
                pool->alive_num -= 1;
                pthread_mutex_unlock(&pool->pool_lock);
                delWorker(pool);
            }
        }
        // 取出任务
        if (pool->isShutdown) {
            pool->alive_num -= 1;
            pthread_mutex_unlock(&pool->pool_lock);
            delWorker(pool);
        }
        Task task;
        task.task_func = pool->task_queue[pool->tq_front].task_func;
        task.task_arg = pool->task_queue[pool->tq_front].task_arg;
        pool->tq_front = (pool->tq_front + 1) % pool->tq_capacity;  // 移动头标
        pool->tq_size -= 1;
        pthread_cond_signal(&pool->tq_notFull_cond);      // 唤醒生产者
        pthread_mutex_unlock(&pool->pool_lock);

        // 执行任务
        printf("task thread %ld start...\n", pthread_self());
        pthread_mutex_lock(&pool->busy_lock);
        pool->busy_num += 1;
        pthread_mutex_unlock(&pool->busy_lock);

        task.task_func(task.task_arg);
        free(task.task_arg);    // task.arg是堆区内存
        task.task_arg = NULL;

        printf("task thread %ld end.\n", pthread_self());
        pthread_mutex_lock(&pool->busy_lock);
        pool->busy_num -= 1;
        pthread_mutex_unlock(&pool->busy_lock);
    }

}


void *manager_job(void *th_pool) {
    /*当线程池开启时，动态添加或销毁线程*/
    ThreadPool *pool = (ThreadPool *)th_pool;
    printf("manager start.\n");
    while (!pool->isShutdown) {
        sleep(2);    // 操作频率

        // 读取参数
        pthread_mutex_lock(&pool->pool_lock);
        int tq_size = pool->tq_size;
        int alive_num = pool->alive_num;
        pthread_mutex_unlock(&pool->pool_lock);

        pthread_mutex_lock(&pool->busy_lock);
        int busy_num = pool->busy_num;
        pthread_mutex_unlock(&pool->busy_lock);

        // 每次操作时线程的增/删个数
        int delta_num = 2;

        // 添加线程（线程数少于任务数）
        if (alive_num < tq_size && alive_num < pool->max_num) {
            pthread_mutex_lock(&pool->pool_lock);
            for (int i = 0, counter = 0;
                 i < pool->max_num && counter < delta_num
                 && pool->alive_num < pool->max_num;
                 ++i){
                if (pool->workersIDs[i] == 0) {
                    pthread_create(&pool->workersIDs[i], NULL, workers_job, pool);
                    counter += 1;
                    pool->alive_num += 1;
                }
            }
            pthread_mutex_unlock(&pool->pool_lock);
        }

        // 销毁线程（线程数远多于正忙线程数）
        if (alive_num > 2*busy_num && alive_num > pool->min_num) {
            pthread_mutex_lock(&pool->pool_lock);
            pool->kill_num = delta_num;
            pthread_mutex_unlock(&pool->pool_lock);

            for (int i = 0; i < delta_num; ++i) {
                pthread_cond_signal(&pool->tq_notEmpty_cond);
            }
        }
    }

    return NULL;
}


// 创建线程池
ThreadPool *creatThreadPool(int min_num, int max_num, int tq_capacity){
    // 开辟线程池
    ThreadPool *pool = (ThreadPool *)malloc(sizeof(ThreadPool));

    do{
        if (pool == NULL) {
            printf("Threads pool malloc failed.\n");
            break;
        }
        // 设置任务队列
        pool->task_queue = (Task *) malloc(tq_capacity * sizeof(Task));
        if (pool->task_queue == NULL) {
            printf("task queue malloc failed.\n");
            break;
        }
        pool->tq_capacity = tq_capacity;
        pool->tq_size = 0;
        pool->tq_front = 0;
        pool->tq_back = 0;

        // 设置线程池
        pool->workersIDs = (pthread_t *) malloc(max_num * sizeof(pthread_t));
        if (pool->workersIDs == NULL) {
            printf(" Workers malloc failed.\n");
            break;
        }
        memset(pool->workersIDs, 0, max_num * sizeof(pthread_t));
        pool->min_num = min_num;
        pool->max_num = max_num;
        pool->busy_num = 0;
        pool->alive_num = min_num;
        pool->kill_num = 0;
        pool->isShutdown = 0;

        if (pthread_mutex_init(&pool->pool_lock, NULL) != 0 ||
            pthread_mutex_init(&pool->busy_lock, NULL) != 0 ||
            pthread_cond_init(&pool->tq_notEmpty_cond, NULL) != 0 ||
            pthread_cond_init(&pool->tq_notFull_cond, NULL) != 0) {
            printf(" mutex or condition init failed.\n");
            break;
        }
        // 创建Manager及Workers线程
        if (pthread_create(&pool->managerID, NULL, manager_job, pool) != 0){
            printf(" manager thread create failed.\n");
            break;
        }

        for (int i = 0; i < min_num; ++i) {
            if (pthread_create(&pool->workersIDs[i], NULL, workers_job, pool) != 0) {
                printf(" worker thread create failed.\n");
                break;
            }
        }

        return pool;

    } while (0);

    // 创建失败时释放堆区空间
    if (pool && pool->workersIDs)  free(pool->workersIDs);
    if (pool && pool->task_queue)  free(pool->task_queue);
    if (pool) free(pool);

    return NULL;
}


// 添加任务
int addTaskToPool(ThreadPool *pool, void (task_func)(void *), void *task_arg) {
    /*任务队列不为空时添加任务， 并唤醒消费者*/
    pthread_mutex_lock(&pool->pool_lock);
    while(pool->tq_size == pool->tq_capacity && !pool->isShutdown){
        // 阻塞等待任务队列有空
        pthread_cond_wait(&pool->tq_notFull_cond, &pool->pool_lock);
    }
    if (pool->isShutdown) {
        pthread_mutex_unlock(&pool->pool_lock);
        return -1;
    }

    // 添加任务
    pool->task_queue[pool->tq_back].task_func = task_func;
    pool->task_queue[pool->tq_back].task_arg = task_arg;
    pool->tq_back = (pool->tq_back + 1) % pool->tq_capacity;
    pool->tq_size += 1;
    pthread_cond_signal(&pool->tq_notEmpty_cond);      // 唤醒消费者

    pthread_mutex_unlock(&pool->pool_lock);
    return 1;
}


// 销毁线程池
int destroyThreadPool(ThreadPool *pool) {
    if (pool == NULL)  return -1;

    // 关闭线程池
    pool->isShutdown = 1;    // 只有本函数会改变shutdown值，无需加锁

    // 阻塞等待Manager线程结束
    pthread_join(pool->managerID, NULL);

    // 结束Workers线程
    while(pool->alive_num > 0) {
        pthread_cond_signal(&pool->tq_notEmpty_cond);
    }

    // 释放内存
    if (pool->task_queue) free(pool->task_queue);
    if (pool->workersIDs) free(pool->workersIDs);
    pthread_mutex_destroy(&pool->pool_lock);
    pthread_mutex_destroy(&pool->busy_lock);
    pthread_cond_destroy(&pool->tq_notFull_cond);
    pthread_cond_destroy(&pool->tq_notEmpty_cond);
    free(pool);
    pool = NULL;
    return 1;
}


// 获取正忙线程个数
int getBusyNum(ThreadPool *pool){
    pthread_mutex_lock(&pool->pool_lock);
    int busy_num = pool->busy_num;
    pthread_mutex_unlock(&pool->pool_lock);
    return busy_num;
}

// 获取线程个数
int getAliveNum(ThreadPool *pool){
    pthread_mutex_lock(&pool->pool_lock);
    int alive_num = pool->alive_num;
    pthread_mutex_unlock(&pool->pool_lock);
    return alive_num;
}



// 设置任务
typedef struct TwoNum{
    int num1;
    int num2;
}TwoNum;
void addTwoNum(void *arg){
    TwoNum *nums = (TwoNum *) arg;
    int result = nums->num1 + nums->num2;
    usleep(result * 80000);
    printf("Worker %ld result: %d\n", pthread_self(), result);
}

void printNum(void *arg){
    int num = *(int *)arg;
    printf("Worker %ld result: %d\n", pthread_self(), num);
    sleep(1);
}


int main() {
    srand((unsigned) time(NULL));
    ThreadPool *thread_pool = creatThreadPool(3, 10, 100);
    if (thread_pool != NULL) {
        for (int i = 0; i < 100; ++i) {
            int * num = (int *) malloc(sizeof(int));
            *num = i + 100;
            addTaskToPool(thread_pool, printNum, num);
        }

        sleep(20);

        for (int i = 0; i < 100; ++i) {
            TwoNum *nums = (TwoNum *) malloc(sizeof(TwoNum));
            nums->num1 = rand() % 10 + 2;
            nums->num2 = rand() % 10 + 2;
            addTaskToPool(thread_pool, addTwoNum, nums);
        }
    }

    sleep(30);
    destroyThreadPool(thread_pool);
    return 0;
}